package com.wgzx.modbus.client;

import com.wgzx.modbus.entity.*;
import com.wgzx.modbus.entity.exception.ConnectionException;
import com.wgzx.modbus.entity.request.*;
import com.wgzx.modbus.entity.response.*;
import com.wgzx.modbus.factory.CodecFactory;
import com.wgzx.modbus.handler.ModbusChannelInitializer;
import com.wgzx.modbus.handler.ModbusResponseHandler;
import com.wgzx.modbus.idgen.IdentifierGenerator;
import com.wgzx.modbus.message.MsgManager;
import com.wgzx.modbus.util.ModbusUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.ScheduledFuture;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

import java.util.BitSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public class ModbusClient {

    private final String host;
    private final int port;
    private final EventLoopGroup eventLoopGroup;
    private final boolean enableLog;
    private final int connectTimeoutMillis;
    private final int retryIntervalMillis;
    private final int protocolIdentifier;
    private final short unitIdentifier;
    private volatile ScheduledFuture<?> retryFuture;
    private final CodecFactory codecFactory;
    private Bootstrap bootstrap;
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final AtomicReference<Channel> channelRef;
    private final MsgManager msgManager;
    private final IdentifierGenerator identifierGenerator;

    public ModbusClient(String host, int port, EventLoopGroup eventLoopGroup, boolean enableLog,
                        int connectTimeoutMillis, int retryIntervalMillis, int protocolIdentifier,
                        short unitIdentifier, CodecFactory codecFactory, MsgManager msgManager,
                        IdentifierGenerator identifierGenerator) {
        this.host = host;
        this.port = port;
        this.eventLoopGroup = eventLoopGroup;
        this.enableLog = enableLog;
        this.connectTimeoutMillis = connectTimeoutMillis;
        this.retryIntervalMillis = retryIntervalMillis;
        this.protocolIdentifier = protocolIdentifier;
        this.unitIdentifier = unitIdentifier;
        this.codecFactory = codecFactory;
        this.msgManager = msgManager;
        this.identifierGenerator = identifierGenerator;
        this.channelRef = new AtomicReference<>();
    }

    /**
     * 开始连接到服务器
     *
     * @throws ConnectionException
     */
    public void connect(boolean async) throws InterruptedException, ConnectionException {
        if (closed.get()) {
            throw new ConnectionException("The client is closed!");
        }

        this.bootstrap = new Bootstrap();
        this.bootstrap
                .group(this.eventLoopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMillis)
                .handler(new ModbusChannelInitializer(codecFactory, new ModbusResponseHandler(msgManager), enableLog));
        this.doConnect(async);
        Runtime.getRuntime().addShutdownHook(new Thread(this::close));
    }

    /**
     * 开始连接服务器
     *
     * @param async 是否异步连接
     */
    private void doConnect(boolean async) throws InterruptedException {
        log.info("Netty客户端尝试与服务器建立连接. 服务器地址:{{}:{}}", host, port);
        ChannelFuture f = this.bootstrap.connect(host, port);
        f.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.info("Netty客户端连接服务器成功. 服务器地址:{{}:{}}", host, port);
                this.addChannelCloseListener(future.channel());
            } else {
                log.warn("Netty客户端连接服务器失败. 服务器地址:{{}:{}}", host, port);
                this.reconnect(future.channel().eventLoop());
            }
        });
        if (!async) f.sync();
    }

    /**
     * 客户端断线重连
     *
     * @param channel
     */
    private void addChannelCloseListener(Channel channel) {
        this.channelRef.set(channel);
        channel.closeFuture().addListener((future) -> {
            log.warn("Netty客户端与服务器连接中断. 服务器地址:{{}:{}}", host, port);
            this.reconnect(channel.eventLoop());
        });
    }

    /**
     * 客户端断线重连
     *
     * @param eventLoop
     */
    private void reconnect(EventLoop eventLoop) {
        if (!closed.get() && !eventLoop.isShutdown()) {
            this.retryFuture = eventLoop.schedule(() -> {
                try {
                    doConnect(true);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }, retryIntervalMillis, TimeUnit.MILLISECONDS);
        }
    }

    /**
     * 关闭客户端连接
     */
    public void close() {
        if (!closed.get()) {
            closed.set(true);
            ModbusUtil.cancelFuture(retryFuture);
            ModbusUtil.closeChannel(channelRef.get());
        }
    }

    /**
     * 客户端是否连接正常
     *
     * @return ture:正常 false:关闭
     */
    private boolean clientIsActive() {
        return !closed.get() && channelRef.get() != null && channelRef.get().isActive();
    }

    /**
     * 构建modbus请求消息
     *
     * @param request modbus请求数据
     * @return ModbusReqFrame
     */
    private ModbusReqFrame<ModbusReq> buildModbusRequest(ModbusReq request) {
        int transactionId = identifierGenerator.calculateTransactionIdentifier();
        int pduLength = request.calculateLength();

        //构建mbap报文头(7字节)   pduLength+1 = pdu消息长度+1字节单元标识符
        MbapHeader header = new MbapHeader(transactionId, protocolIdentifier, pduLength + 1, unitIdentifier);
        ModbusReqFrame<ModbusReq> frame = new ModbusReqFrame<>(header, request);
        return frame;
    }

    /**
     * 发送modbus请求消息
     *
     * @param frame modbus请求消息
     * @throws ConnectionException
     */
    private void sendRequest(ModbusReqFrame frame) throws ConnectionException {
        if (!clientIsActive()) {
            throw new ConnectionException("Client not connected!");
        }
        channelRef.get().writeAndFlush(frame);
    }

    /**
     * 异步读线圈
     *
     * @param startAddress    起始地址(0x0000至0xFFFF)
     * @param quantityOfCoils 线圈数量(0x01至0x07D0)
     * @return Mono<ReadCoilsResponse>
     * @throws ConnectionException
     */
    public Mono<ReadCoilsResponse> readCoilsAsync(int startAddress, int quantityOfCoils) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadCoilsRequest(startAddress, quantityOfCoils));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步读离散量输入
     *
     * @param startAddress       起始地址(0x0000至0xFFFF)
     * @param quantityOfDiscrete 输入数量(0x01至0x07D0)
     * @return Mono<ReadDiscreteInputsResponse>
     * @throws ConnectionException
     */
    public Mono<ReadDiscreteInputsResponse> readDiscreteInputsAsync(int startAddress, int quantityOfDiscrete) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadDiscreteInputsRequest(startAddress, quantityOfDiscrete));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步读保持寄存器
     *
     * @param startAddress             起始地址(0x0000至0xFFFF)
     * @param quantityOfInputRegisters 寄存器数量(0x01至0x7D)
     * @return Mono<ReadHoldingRegistersResponse>
     * @throws ConnectionException
     */
    public Mono<ReadHoldingRegistersResponse> readHoldingRegistersRequestAsync(int startAddress, int quantityOfInputRegisters) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadHoldingRegistersRequest(startAddress, quantityOfInputRegisters));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步读输入寄存器
     *
     * @param startAddress             起始地址(0x0000至0xFFFF)
     * @param quantityOfInputRegisters 寄存器数量(0x01至0x007D)
     * @return Mono<ReadInputRegistersResponse>
     * @throws ConnectionException
     */
    public Mono<ReadInputRegistersResponse> readInputRegistersRequestAsync(int startAddress, int quantityOfInputRegisters) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadInputRegistersRequest(startAddress, quantityOfInputRegisters));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步写单个线圈
     *
     * @param outputAddress 输出地址(0x0000至0xFFFF)
     * @param state         输出值
     * @return Mono<WriteSingleCoilResponse>
     * @throws ConnectionException
     */
    public Mono<WriteSingleCoilResponse> writeSingleCoilRequestAsync(int outputAddress, boolean state) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteSingleCoilRequest(outputAddress, state));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步写单个寄存器
     *
     * @param address 寄存器地址(0x0000至0xFFFF)
     * @param value   寄存器值(0x0000至0xFFFF)
     * @return Mono<WriteSingleRegisterResponse>
     * @throws ConnectionException
     */
    public Mono<WriteSingleRegisterResponse> writeSingleRegisterRequestAsync(int address, int value) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteSingleRegisterRequest(address, value));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步写多个线圈
     *
     * @param startingAddress   寄存器地址(0x0000至0xFFFF)
     * @param quantityOfOutputs 输出数量(0x0001至0x07B0)
     * @param outputsValue      输出值(bit位图)
     * @return Mono<WriteSingleRegisterResponse>
     * @throws ConnectionException
     */
    public Mono<WriteMultipleCoilsResponse> writeMultipleCoilsRequestAsync(int startingAddress, int quantityOfOutputs, BitSet outputsValue) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteMultipleCoilsRequest(startingAddress, quantityOfOutputs, outputsValue));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 异步写多个寄存器
     *
     * @param startingAddress 寄存器地址(0x0000至0xFFFF)
     * @param registers       寄存器值
     * @return Mono<WriteMultipleRegistersResponse>
     * @throws ConnectionException
     */
    public Mono<WriteMultipleRegistersResponse> writeMultipleRegistersRequestAsync(int startingAddress, int[] registers) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteMultipleRegistersRequest(startingAddress, registers));
        return msgManager.getResponseAsync(request.getHeader().getTransactionIdentifier(), () -> sendRequest(request));
    }

    /**
     * 同步读线圈
     *
     * @param startAddress    起始地址(0x0000至0xFFFF)
     * @param quantityOfCoils 线圈数量(0x01至0x07D0)
     * @return CompletableFuture<ReadCoilsResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<ReadCoilsResponse> readCoils(int startAddress, int quantityOfCoils) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadCoilsRequest(startAddress, quantityOfCoils));
        CompletableFuture<ReadCoilsResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步读离散量输入
     *
     * @param startAddress       起始地址(0x0000至0xFFFF)
     * @param quantityOfDiscrete 输入数量(0x01至0x07D0)
     * @return CompletableFuture<ReadDiscreteInputsResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<ReadDiscreteInputsResponse> readDiscreteInputs(int startAddress, int quantityOfDiscrete) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadDiscreteInputsRequest(startAddress, quantityOfDiscrete));
        CompletableFuture<ReadDiscreteInputsResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步读保持寄存器
     *
     * @param startAddress             起始地址(0x0000至0xFFFF)
     * @param quantityOfInputRegisters 寄存器数量(0x01至0x7D)
     * @return CompletableFuture<ReadHoldingRegistersResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<ReadHoldingRegistersResponse> readHoldingRegistersRequest(int startAddress, int quantityOfInputRegisters) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadHoldingRegistersRequest(startAddress, quantityOfInputRegisters));
        CompletableFuture<ReadHoldingRegistersResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步读输入寄存器
     *
     * @param startAddress             起始地址(0x0000至0xFFFF)
     * @param quantityOfInputRegisters 寄存器数量(0x01至0x007D)
     * @return CompletableFuture<ReadInputRegistersResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<ReadInputRegistersResponse> readInputRegistersRequest(int startAddress, int quantityOfInputRegisters) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new ReadInputRegistersRequest(startAddress, quantityOfInputRegisters));
        CompletableFuture<ReadInputRegistersResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步写单个线圈
     *
     * @param outputAddress 输出地址(0x0000至0xFFFF)
     * @param state         输出值
     * @return CompletableFuture<WriteSingleCoilResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<WriteSingleCoilResponse> writeSingleCoilRequest(int outputAddress, boolean state) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteSingleCoilRequest(outputAddress, state));
        CompletableFuture<WriteSingleCoilResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步写单个寄存器
     *
     * @param address 寄存器地址(0x0000至0xFFFF)
     * @param value   寄存器值(0x0000至0xFFFF)
     * @return CompletableFuture<WriteSingleRegisterResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<WriteSingleRegisterResponse> writeSingleRegisterRequest(int address, int value) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteSingleRegisterRequest(address, value));
        CompletableFuture<WriteSingleRegisterResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步写多个线圈
     *
     * @param startingAddress   寄存器地址(0x0000至0xFFFF)
     * @param quantityOfOutputs 输出数量(0x0001至0x07B0)
     * @param outputsValue      输出值(bit位图)
     * @return CompletableFuture<WriteSingleRegisterResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<WriteMultipleCoilsResponse> writeMultipleCoilsRequest(int startingAddress, int quantityOfOutputs, BitSet outputsValue) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteMultipleCoilsRequest(startingAddress, quantityOfOutputs, outputsValue));
        CompletableFuture<WriteMultipleCoilsResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }

    /**
     * 同步写多个寄存器
     *
     * @param startingAddress 寄存器地址(0x0000至0xFFFF)
     * @param registers       寄存器值
     * @return Mono<WriteMultipleRegistersResponse>
     * @throws ConnectionException
     */
    public CompletableFuture<WriteMultipleRegistersResponse> writeMultipleRegistersRequest(int startingAddress, int[] registers) throws ConnectionException {
        ModbusReqFrame request = buildModbusRequest(new WriteMultipleRegistersRequest(startingAddress, registers));
        CompletableFuture<WriteMultipleRegistersResponse> future = msgManager.getResponseSync(request.getHeader().getTransactionIdentifier());
        this.sendRequest(request);
        return future;
    }
}
